package cn.tpshion.pulsar.support;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.util.CollectionUtils;

import java.util.Set;
import java.util.concurrent.TimeUnit;

public class ConsumerFactoryBean implements FactoryBean<Consumer> {

    private PulsarClient pulsarClient;

    private PulsarConsumer consumer;

    private MessageListener defaultMessageListener;

    public ConsumerFactoryBean(PulsarClient pulsarClient, PulsarConsumer consumer, MessageListener messageListener) {
        this.pulsarClient = pulsarClient;
        this.consumer = consumer;
        this.defaultMessageListener = messageListener;
    }

    @Override
    public Consumer getObject() throws Exception {
        ConsumerBuilder builder = pulsarClient.newConsumer(consumer.getSchema());
        builder.consumerName(consumer.getConsumerName());
        builder.subscriptionName(consumer.getSubscriptionName());
        builder.subscriptionType(consumer.getSubscriptionType());
        builder.receiverQueueSize(consumer.getReceiverQueueSize());
        builder.ackTimeoutTickTime(consumer.getAckTimeoutMillis(), TimeUnit.MILLISECONDS);
        Set<String> topics = consumer.getTopics();
        if(CollectionUtils.isEmpty(topics)){
            builder.topic(consumer.buildTopic());
        }else{
            builder.topics(consumer.buildTopics());
        }
        builder.messageListener(defaultMessageListener);
        return builder.subscribe();
    }

    @Override
    public Class<?> getObjectType() {
        return Consumer.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

}
